package tasks.io.client;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import tasks.io.core.*;
import tasks.io.core.lock.SetWithLock;
import tasks.io.core.stat.ChannelStat;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.StandardSocketOptions;
import java.nio.channels.AsynchronousChannelGroup;
import java.nio.channels.AsynchronousSocketChannel;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantReadWriteLock.ReadLock;


public class Client {
    private static final Logger log = LoggerFactory.getLogger(Client.class);

    private final AsynchronousChannelGroup asynchronousChannelGroup;

    private final TioClientConfig tioClientConfig;


    public Client(final TioClientConfig tioClientConfig) throws IOException {
        super();
        this.tioClientConfig = tioClientConfig;
        this.asynchronousChannelGroup = AsynchronousChannelGroup.withThreadPool(tioClientConfig.groupExecutor);

        startHeartbeatTask();
        startReconnTask();
    }


    public void asynConnect(Node serverNode) throws Exception {
        asynConnect(serverNode, null);
    }


    public void asynConnect(Node serverNode, Integer timeout) throws Exception {
        asynConnect(serverNode, null, null, timeout);
    }

    public void asynConnect(Node serverNode, String bindIp, Integer bindPort, Integer timeout) throws Exception {
        connect(bindIp, bindPort, serverNode, null, timeout, false);
    }


    public ClientChannelContext connect(Node serverNode) throws Exception {
        return connect(serverNode, null);
    }


    public ClientChannelContext connect(Node serverNode, Integer timeout) throws Exception {
        return connect(serverNode, null, 0, timeout);
    }

    /**
     * @param serverNode
     * @param bindIp
     * @param bindPort
     * @param initClientChannelContext
     * @param timeout                  超时时间，单位秒
     * @return
     * @throws Exception
     * @author tanyaowu
     */
    public ClientChannelContext connect(Node serverNode, String bindIp, Integer bindPort, ClientChannelContext initClientChannelContext, Integer timeout) throws Exception {
        return connect(bindIp, bindPort, serverNode, initClientChannelContext, timeout, true);
    }

    /**
     * @param bindIp
     * @param bindPort
     * @param serverNode
     * @param initClientChannelContext
     * @param timeout                  超时时间，单位秒
     * @param isSyn                    true: 同步, false: 异步
     * @return
     * @throws Exception
     * @author tanyaowu
     */
    private ClientChannelContext connect(String bindIp, Integer bindPort, Node serverNode, ClientChannelContext initClientChannelContext, Integer timeout, boolean isSyn)
            throws Exception {


        AsynchronousSocketChannel asynchronousSocketChannel = AsynchronousSocketChannel.open(asynchronousChannelGroup);
        //
        asynchronousSocketChannel.setOption(StandardSocketOptions.TCP_NODELAY, true);
        asynchronousSocketChannel.setOption(StandardSocketOptions.SO_REUSEADDR, true);
        asynchronousSocketChannel.setOption(StandardSocketOptions.SO_KEEPALIVE, true);

        InetSocketAddress bind = null;
        if (bindPort != null && bindPort > 0) {
            if (bindIp != null) {
                bind = new InetSocketAddress(bindIp, bindPort);
            } else {
                bind = new InetSocketAddress(bindPort);
            }
        }

        if (bind != null) {
            asynchronousSocketChannel.bind(bind);
        }


        boolean isReconnect = initClientChannelContext != null;
        ClientChannelContext clientChannelContext = initClientChannelContext;


        InetSocketAddress inetSocketAddress = new InetSocketAddress(serverNode.getIp(), serverNode.getPort());

        ConnectionCompletionHandlerVo attachment = new ConnectionCompletionHandlerVo(this, bindIp, bindPort, serverNode, asynchronousSocketChannel, clientChannelContext, isReconnect);
        ConnectionCompletionHandler connectionCompletionHandler = new ConnectionCompletionHandler();
        if (isSyn) {
            Integer realTimeout = timeout;
            if (realTimeout == null) {
                realTimeout = 5;
            }

            CountDownLatch countDownLatch = new CountDownLatch(1);
            attachment.setCountDownLatch(countDownLatch);

            try {
                asynchronousSocketChannel.connect(inetSocketAddress, attachment, connectionCompletionHandler);
            } catch (Throwable e) {
                connectionCompletionHandler.failed(e, attachment);
                return attachment.getClientChannelContext();
            }


            countDownLatch.await(realTimeout, TimeUnit.SECONDS);
            return attachment.getClientChannelContext();
        } else {
            try {
                asynchronousSocketChannel.connect(inetSocketAddress, attachment, connectionCompletionHandler);
            } catch (Throwable e) {
                connectionCompletionHandler.failed(e, attachment);
            }
            return null;
        }
    }

    /**
     * @param serverNode
     * @param bindIp
     * @param bindPort
     * @param timeout    超时时间，单位秒
     * @return
     * @throws Exception
     * @author tanyaowu
     */
    public ClientChannelContext connect(Node serverNode, String bindIp, Integer bindPort, Integer timeout) throws Exception {
        return connect(serverNode, bindIp, bindPort, null, timeout);
    }

    /**
     * @return the tioClientConfig
     */
    public TioClientConfig getTioClientConfig() {
        return tioClientConfig;
    }

    /**
     * @param channelContext
     * @param timeout        单位秒
     * @return
     * @throws Exception
     * @author tanyaowu
     */
    public void reconnect(ClientChannelContext channelContext, Integer timeout) throws Exception {
        connect(channelContext.getServerNode(), channelContext.getBindIp(), channelContext.getBindPort(), channelContext, timeout);
    }

    /**
     * 定时任务：发心跳
     */
    private void startHeartbeatTask() {
        final ClientGroupStat clientGroupStat = (ClientGroupStat) tioClientConfig.groupStat;
        final TioClientHandler tioHandler = tioClientConfig.getTioClientHandler();

        final String id = tioClientConfig.getId();
        new Thread(new Runnable() {
            @Override
            public void run() {
                while (!tioClientConfig.isStopped()) {
                    //					final long heartbeatTimeout = tioClientConfig.heartbeatTimeout;
                    if (tioClientConfig.heartbeatTimeout <= 0) {
                        log.warn("用户取消了框架层面的心跳定时发送功能，请用户自己去完成心跳机制");
                        break;
                    }
                    SetWithLock<ChannelContext> setWithLock = tioClientConfig.connecteds;
                    ReadLock readLock = setWithLock.readLock();
                    readLock.lock();
                    try {
                        Set<ChannelContext> set = setWithLock.getObj();
                        long currtime = SystemTimer.currTime;
                        for (ChannelContext channelContext1 : set) {
                            ClientChannelContext clientChannelContext = (ClientChannelContext) channelContext1;
                            if (clientChannelContext.isClosed || clientChannelContext.isRemoved) {
                                continue;
                            }

                            ChannelStat channelStat = clientChannelContext.channelStat;
                            long compareTime = Math.max(channelStat.latestTimeOfReceivedByte, channelStat.latestTimeOfSentPacket);
                            long interval = currtime - compareTime;
                            if (interval >= tioClientConfig.heartbeatTimeout / 2) {
                                Packet packet = tioHandler.heartbeatPacket(clientChannelContext);
                                if (packet != null) {
                                    if (log.isInfoEnabled()) {
                                        log.info("{}发送心跳包", clientChannelContext);
                                    }
                                    Io.send(clientChannelContext, packet);
                                }
                            }
                        }
                        if (log.isInfoEnabled()) {
                            log.info("[{}]: curr:{}, closed:{}, received:({}p)({}b), handled:{}, sent:({}p)({}b)", id, set.size(), clientGroupStat.closed.get(),
                                    clientGroupStat.receivedPackets.get(), clientGroupStat.receivedBytes.get(), clientGroupStat.handledPackets.get(),
                                    clientGroupStat.sentPackets.get(), clientGroupStat.sentBytes.get());
                        }

                    } catch (Throwable e) {
                        log.error("", e);
                    } finally {
                        try {
                            readLock.unlock();
                            Thread.sleep(tioClientConfig.heartbeatTimeout / 4);
                        } catch (Throwable e) {
                            log.error(e.toString(), e);
                        } finally {

                        }
                    }
                }
            }
        }, "tio-timer-heartbeat" + id).start();
    }

    /**
     * 启动重连任务
     *
     * @author tanyaowu
     */
    private void startReconnTask() {
        final ReconnConf reconnConf = tioClientConfig.getReconnConf();
        if (reconnConf == null || reconnConf.getInterval() <= 0) {
            return;
        }

        final String id = tioClientConfig.getId();
        Thread thread = new Thread(new Runnable() {
            @Override
            public void run() {
                while (!tioClientConfig.isStopped()) {
                    log.error("closeds:{}, connections:{}", tioClientConfig.closeds.size(), tioClientConfig.connections.size());
                    //log.info("准备重连");
                    LinkedBlockingQueue<ChannelContext> queue = reconnConf.getQueue();
                    ClientChannelContext clientChannelContext = null;
                    try {
                        clientChannelContext = (ClientChannelContext) queue.take();
                    } catch (InterruptedException e1) {
                        log.error(e1.toString(), e1);
                    }
                    if (clientChannelContext == null) {
                        continue;
                        //						return;
                    }

                    if (clientChannelContext.isRemoved) //已经删除的，不需要重新再连
                    {
                        continue;
                    }


                    long sleeptime = reconnConf.getInterval() - (SystemTimer.currTime - clientChannelContext.channelStat.timeInReconnQueue);
                    //log.info("sleeptime:{}, closetime:{}", sleeptime, timeInReconnQueue);
                    if (sleeptime > 0) {
                        try {
                            Thread.sleep(sleeptime);
                        } catch (InterruptedException e) {
                            log.error(e.toString(), e);
                        }
                    }

                    if (clientChannelContext.isRemoved || !clientChannelContext.isClosed) //已经删除的和已经连上的，不需要重新再连
                    {
                        continue;
                    } else {
                        ReconnTask runnable = clientChannelContext.getReconnRunnable();
                        if (runnable == null) {
                            synchronized (clientChannelContext) {
                                runnable = clientChannelContext.getReconnRunnable();
                                if (runnable == null) {
                                    runnable = new ReconnTask(clientChannelContext, Client.this, reconnConf.getThreadPoolExecutor());
                                    clientChannelContext.setReconnRunnable(runnable);
                                }
                            }
                        }
                        runnable.execute();
                        //						reconnConf.getThreadPoolExecutor().execute(runnable);
                    }
                }
            }
        });
        thread.setName("tio-timer-reconnect-" + id);
        thread.setDaemon(true);
        thread.start();

    }

    /**
     * @return
     * @author tanyaowu
     */
    public boolean stop() {
        boolean ret = true;
        try {
            tioClientConfig.groupExecutor.shutdown();
        } catch (Exception e1) {
            log.error(e1.toString(), e1);
        }
        try {
            tioClientConfig.tioExecutor.shutdown();
        } catch (Exception e1) {
            log.error(e1.toString(), e1);
        }

        tioClientConfig.setStopped(true);
        try {
            ret = ret && tioClientConfig.groupExecutor.awaitTermination(6000, TimeUnit.SECONDS);
            ret = ret && tioClientConfig.tioExecutor.awaitTermination(6000, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            log.error(e.getLocalizedMessage(), e);
        }
        log.info("client resource has released");
        return ret;
    }
}
